package com.hub.realtime.resource.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.hub.realtime.common.core.domain.model.ClusterInfo;
import com.hub.realtime.common.core.domain.model.LoginUser;
import com.hub.realtime.common.exception.UtilException;
import com.hub.realtime.common.utils.DateUtils;
import com.hub.realtime.common.utils.StringUtils;
import com.hub.realtime.framework.service.ServiceBase;
import com.hub.realtime.resource.domain.ResCluster;
import com.hub.realtime.resource.domain.ResFlinkVersions;
import com.hub.realtime.resource.dto.input.ApplicationQueryParam;
import com.hub.realtime.resource.service.IResClusterService;
import com.hub.realtime.resource.service.IResFlinkVersionsService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.hub.realtime.resource.mapper.ResApplicationMapper;
import com.hub.realtime.resource.domain.ResApplication;
import com.hub.realtime.resource.service.IResApplicationService;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

/**
 * 项目管理表Service业务层处理
 *
 * @author wupeng
 * @date 2022-01-05
 */
@Service
public class ResApplicationServiceImpl extends ServiceBase<ResApplicationMapper, ResApplication> implements IResApplicationService {


    @Autowired
    IResClusterService clusterService;

    @Autowired
    IResFlinkVersionsService flinkVersionsService;

    /**
     * 分页查询
     *
     * @param param
     * @return
     */
    @Override
    public Page<ResApplication> list(ApplicationQueryParam param) {
        LambdaQueryWrapper<ResApplication> wrapper = createCommonWrapper(ResApplication::getUserId, ResApplication::getCreateTime);
        if (StringUtils.isNotEmpty(param.getName())) {
            wrapper.like(ResApplication::getName, param.getName());
        }
        return getPageList(param, wrapper);
    }

    /**
     * 保存和新增
     *
     * @param application
     * @return
     */
    @Override
    public int saveInfo(ResApplication application) {
        int res = -1;
        LoginUser loginUser = getLoginUser();
        LambdaQueryWrapper<ResApplication> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(ResApplication::getName, application.getName());
        ResApplication resApplication = baseMapper.selectOne(queryWrapper);
        if (resApplication != null) {
            if (application.getId() != null) {
                if (!resApplication.getId().equals(application.getId())) {
                    throw new UtilException("存在相同名称的项目！");
                } else {
                    application.setUpdateTime(DateUtils.getNowDate());
                    application.setUpdateBy(loginUser.getUsername());
                    res = baseMapper.updateById(application);
                }
            } else {
                throw new UtilException("存在相同名称的项目！");
            }
        } else {
            if (application.getId() == null) {
                application.setUid(UUID.randomUUID().toString());
                application.setUserId(loginUser.getUserId());
                application.setCreateBy(loginUser.getUsername());
                application.setUserUid(getCurrentUser(null).getUid());
                application.setCreateTime(DateUtils.getNowDate());
                res = baseMapper.insert(application);
            } else {
                application.setUpdateTime(DateUtils.getNowDate());
                application.setUpdateBy(loginUser.getUsername());
                res = baseMapper.updateById(application);
            }
        }
        updateClusterInfo();
        return res;
    }

    @Override
    public int deleteInfos(Long[] ids) {
        int res = delByIds(ids);
        updateClusterInfo();
        return res;
    }

    @Override
    public List<ResApplication> getAppList() {
        LambdaQueryWrapper<ResApplication> wrapper = createCommonWrapper(ResApplication::getUserId, ResApplication::getCreateTime);
        return baseMapper.selectList(wrapper);
    }

    /**
     * 根据项目ID获取集群信息
     *
     * @param appId
     * @return
     */
    @Override
    public ClusterInfo getClusterInfoByAppId(Long appId) {
        ResApplication application = baseMapper.selectById(appId);
        ResCluster cluster = clusterService.getBaseMapper().selectById(application.getClusterId());
        ClusterInfo clusterInfo = new ClusterInfo();
        clusterInfo.setKerberosConfDir(cluster.getKererosDir());
        clusterInfo.setHadoopConfDir(cluster.getHadoopDir());
        return clusterInfo;
    }

    @Override
    public String getCurrentFlinkVersion(Long appId) {
        ResApplication application = baseMapper.selectById(appId);
        ResCluster cluster = clusterService.getBaseMapper().selectById(application.getClusterId());
        return cluster.getFlinkVersion();
    }

    @Override
    public String getFlinkHome(Long appId) {
        ResApplication application = baseMapper.selectById(appId);
        ResCluster cluster = clusterService.getBaseMapper().selectById(application.getClusterId());
        String flinkVersion = cluster.getFlinkVersion();
        LambdaQueryWrapper<ResFlinkVersions>flinkVersionsLambdaQueryWrapper=new LambdaQueryWrapper<>();
        flinkVersionsLambdaQueryWrapper.eq(ResFlinkVersions::getFlinkVersion,flinkVersion);
        ResFlinkVersions flinkVersions = flinkVersionsService.getBaseMapper().selectOne(flinkVersionsLambdaQueryWrapper);
        return flinkVersions.getLocalPath();
    }


    /**
     * 更新集群信息
     */
    private void updateClusterInfo() {
        LambdaQueryWrapper<ResCluster> queryClusterWrapper = new LambdaQueryWrapper<>();
        LambdaQueryWrapper<ResApplication> otherWrap = new LambdaQueryWrapper<>();
        List<ResApplication> resApplications = baseMapper.selectList(otherWrap);
        if (!resApplications.isEmpty()) {
            //当前使用的集群
            List<Long> clusterIds = new ArrayList<>();
            resApplications.forEach(app -> {
                Long clusterId = app.getClusterId();
                clusterIds.add(clusterId);
            });
            if (!clusterIds.isEmpty()) {
                List<ResCluster> resClusters = clusterService.getBaseMapper().selectBatchIds(clusterIds);
                if (resClusters != null && !resClusters.isEmpty()) {
                    resClusters.forEach(d -> {
                        d.setIsInUse("true");
                        clusterService.getBaseMapper().updateById(d);
                    });
                }
            }
            //把不属于当前项目的所有集群都设置为没有在使用
            queryClusterWrapper.notIn(ResCluster::getId, clusterIds);
            List<ResCluster> notInUseClusterList = clusterService.getBaseMapper().selectList(queryClusterWrapper);
            if (notInUseClusterList != null && !notInUseClusterList.isEmpty()) {
                notInUseClusterList.forEach(d -> {
                    d.setIsInUse("false");
                    clusterService.getBaseMapper().updateById(d);
                });
            }
        } else {
            List<ResCluster> resClusters = clusterService.getBaseMapper().selectList(queryClusterWrapper);
            if (resClusters != null && !resClusters.isEmpty()) {
                resClusters.forEach(d -> {
                    d.setIsInUse("false");
                    clusterService.getBaseMapper().updateById(d);
                });
            }
        }
    }


}

